package com.qs.rocketmq.sc.listener;

import com.alibaba.fastjson.JSON;
import com.qs.rocketmq.sc.controller.ScMQTestController;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;

/**
 * 创建事务监听器
 * 这里声明监听生产组为sc_producer_group的所有生产者发送的消息
 *
 * @author QinYupeng
 * @since 2021-9-19 17:14:36
 */
@RocketMQTransactionListener
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private static final Logger logger = LoggerFactory.getLogger(TransactionListenerImpl.class);

    /**
     * 在调用这个方法之前，已经向RocketMQ发送了一条事务消息，之后会调用这个方法，根据这个方法返回的本地事务状态，来决定事务消息提交或者回滚
     *
     * @param msg Spring封装的消息对象
     * @param arg
     * @return 返回本地事务状态
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 从消息 Header 中解析到 args 参数，并使用 JSON 反序列化
        ScMQTestController.Args args = JSON.parseObject(msg.getHeaders().get("args", String.class),
                ScMQTestController.Args.class);
        logger.info("[executeLocalTransaction][执行本地事务，消息：{} args：{}]", msg, args);
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    /**
     * 消息回查
     * 在事务消息长事件未被提交或回滚时，RocketMQ 会回查事务消息对应的生产者分组下的 Producer ，获得事务消息的状态。此时，该方法就会被调用。
     *
     * @param msg Spring封装的消息对象
     * @return 返回本地事务状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        logger.info("[checkLocalTransaction][回查消息：{}]", msg);
        return RocketMQLocalTransactionState.COMMIT;
    }
}
